#include <algorithm>
#include <iostream>
#include <map>
#include <memory>
#include <optional>
#include <string>
#include <utility>
#include <vector>

#include "base_test.hpp"

#include "expression/expression_functional.hpp"
#include "operators/abstract_read_only_operator.hpp"
#include "operators/limit.hpp"
#include "operators/print.hpp"
#include "operators/projection.hpp"
#include "operators/table_scan.hpp"
#include "operators/table_scan/column_between_table_scan_impl.hpp"
#include "operators/table_scan/column_is_null_table_scan_impl.hpp"
#include "operators/table_scan/column_like_table_scan_impl.hpp"
#include "operators/table_scan/column_vs_column_table_scan_impl.hpp"
#include "operators/table_scan/column_vs_value_table_scan_impl.hpp"
#include "operators/table_scan/expression_evaluator_table_scan_impl.hpp"
#include "operators/table_wrapper.hpp"
#include "storage/chunk_encoder.hpp"
#include "storage/encoding_type.hpp"
#include "storage/reference_segment.hpp"
#include "storage/table.hpp"
#include "types.hpp"
#include "utils/assert.hpp"

namespace hyrise {

using namespace expression_functional;  // NOLINT(build/namespaces)

class OperatorsTableScanTest : public BaseTest, public ::testing::WithParamInterface<EncodingType> {
 protected:
  void SetUp() override {
    _encoding_type = GetParam();

    auto int_int_7 = load_table("resources/test_data/tbl/int_int_shuffled.tbl", ChunkOffset{7});
    auto int_int_5 = load_table("resources/test_data/tbl/int_int_shuffled_2.tbl", ChunkOffset{5});

    ChunkEncoder::encode_chunks(int_int_7, {ChunkID{0}, ChunkID{1}}, SegmentEncodingSpec{_encoding_type});
    // partly compressed table
    ChunkEncoder::encode_chunks(int_int_5, {ChunkID{0}, ChunkID{1}}, SegmentEncodingSpec{_encoding_type});

    _int_int_compressed = std::make_shared<TableWrapper>(std::move(int_int_7));
    _int_int_compressed->never_clear_output();
    _int_int_compressed->execute();
    _int_int_partly_compressed = std::make_shared<TableWrapper>(std::move(int_int_5));
    _int_int_partly_compressed->never_clear_output();
    _int_int_partly_compressed->execute();
  }

  std::shared_ptr<TableWrapper> load_and_encode_table(
      const std::string& path, const ChunkOffset chunk_size = ChunkOffset{2},
      const std::optional<std::vector<SortColumnDefinition>> sorted_by = std::nullopt) {
    const auto table = load_table(path, chunk_size);

    auto chunk_encoding_spec = ChunkEncodingSpec{};
    for (const auto& column_definition : table->column_definitions()) {
      if (encoding_supports_data_type(_encoding_type, column_definition.data_type)) {
        chunk_encoding_spec.emplace_back(_encoding_type);
      } else {
        chunk_encoding_spec.emplace_back(EncodingType::Unencoded);
      }
    }

    ChunkEncoder::encode_all_chunks(table, chunk_encoding_spec);

    if (sorted_by) {
      const auto chunk_count = table->chunk_count();
      for (auto chunk_id = ChunkID{0}; chunk_id < chunk_count; ++chunk_id) {
        const auto chunk = table->get_chunk(chunk_id);
        if (!chunk) {
          continue;
        }

        chunk->set_individually_sorted_by(*sorted_by);
      }
    }

    auto table_wrapper = std::make_shared<TableWrapper>(table);
    table_wrapper->never_clear_output();
    table_wrapper->execute();

    return table_wrapper;
  }

  std::shared_ptr<TableWrapper> get_int_float_op() {
    return load_and_encode_table("resources/test_data/tbl/int_float.tbl");
  }

  std::shared_ptr<TableWrapper> get_int_sorted_op() {
    return load_and_encode_table(
        "resources/test_data/tbl/int_sorted.tbl", ChunkOffset{4},
        std::make_optional(std::vector<SortColumnDefinition>{SortColumnDefinition(ColumnID(0), SortMode::Ascending)}));
  }

  std::shared_ptr<TableWrapper> get_int_only_null_op() {
    return load_and_encode_table(
        "resources/test_data/tbl/int_only_null.tbl", ChunkOffset{4},
        std::make_optional(std::vector<SortColumnDefinition>{SortColumnDefinition(ColumnID(0), SortMode::Ascending)}));
  }

  std::shared_ptr<TableWrapper> get_int_string_op() {
    return load_and_encode_table("resources/test_data/tbl/int_string.tbl");
  }

  std::shared_ptr<TableWrapper> get_int_float_with_null_op(const ChunkOffset chunk_size = ChunkOffset{2}) {
    return load_and_encode_table("resources/test_data/tbl/int_float_with_null.tbl", chunk_size);
  }

  std::shared_ptr<TableWrapper> get_table_op_filtered() {
    TableColumnDefinitions table_column_definitions;
    table_column_definitions.emplace_back("a", DataType::Int, false);
    table_column_definitions.emplace_back("b", DataType::Int, false);

    const auto table = std::make_shared<Table>(table_column_definitions, TableType::References);

    const auto test_table_part_compressed = _int_int_partly_compressed->get_output();

    auto pos_list = std::make_shared<RowIDPosList>();
    pos_list->emplace_back(ChunkID{2}, ChunkOffset{0});
    pos_list->emplace_back(ChunkID{1}, ChunkOffset{1});
    pos_list->emplace_back(ChunkID{1}, ChunkOffset{3});
    pos_list->emplace_back(ChunkID{0}, ChunkOffset{2});
    pos_list->emplace_back(ChunkID{2}, ChunkOffset{2});
    pos_list->emplace_back(ChunkID{0}, ChunkOffset{0});
    pos_list->emplace_back(ChunkID{0}, ChunkOffset{4});

    auto segment_a = std::make_shared<ReferenceSegment>(test_table_part_compressed, ColumnID{0}, pos_list);
    auto segment_b = std::make_shared<ReferenceSegment>(test_table_part_compressed, ColumnID{1}, pos_list);

    const auto segments = Segments({segment_a, segment_b});

    table->append_chunk(segments);
    auto table_wrapper = std::make_shared<TableWrapper>(std::move(table));
    table_wrapper->never_clear_output();
    table_wrapper->execute();

    return table_wrapper;
  }

  std::shared_ptr<TableWrapper> get_table_op_with_n_dict_entries(const int num_entries) {
    // Set up dictionary encoded table with a dictionary consisting of num_entries entries.
    auto table_column_definitions = TableColumnDefinitions{};
    table_column_definitions.emplace_back("a", DataType::Int, false);

    const auto table = std::make_shared<Table>(table_column_definitions, TableType::Data, ChunkOffset{100'000});

    for (int i = 0; i <= num_entries; i++) {
      table->append({i});
    }

    table->get_chunk(static_cast<ChunkID>(ChunkID{0}))->finalize();

    ChunkEncoder::encode_chunks(table, {ChunkID{0}}, SegmentEncodingSpec{_encoding_type});

    auto table_wrapper = std::make_shared<TableWrapper>(std::move(table));
    table_wrapper->never_clear_output();
    table_wrapper->execute();

    return table_wrapper;
  }

  std::shared_ptr<const Table> create_referencing_table_w_null_row_id(const bool references_dict_segment) {
    const auto table = load_table("resources/test_data/tbl/int_int_w_null_8_rows.tbl", ChunkOffset{4});

    if (references_dict_segment) {
      ChunkEncoder::encode_all_chunks(table, SegmentEncodingSpec{_encoding_type});
    }

    auto pos_list_a = std::make_shared<RowIDPosList>(
        RowIDPosList{RowID{ChunkID{0}, ChunkOffset{1}}, RowID{ChunkID{1}, ChunkOffset{0}},
                     RowID{ChunkID{0}, ChunkOffset{2}}, RowID{ChunkID{0}, ChunkOffset{3}}});
    auto ref_segment_a = std::make_shared<ReferenceSegment>(table, ColumnID{0}, pos_list_a);

    auto pos_list_b = std::make_shared<RowIDPosList>(RowIDPosList{NULL_ROW_ID, RowID{ChunkID{0}, ChunkOffset{0}},
                                                                  RowID{ChunkID{1}, ChunkOffset{2}},
                                                                  RowID{ChunkID{0}, ChunkOffset{1}}});
    auto ref_segment_b = std::make_shared<ReferenceSegment>(table, ColumnID{1}, pos_list_b);

    TableColumnDefinitions column_definitions;
    column_definitions.emplace_back("a", DataType::Int, true);
    column_definitions.emplace_back("b", DataType::Int, true);
    auto ref_table = std::make_shared<Table>(column_definitions, TableType::References);

    const auto segments = Segments({ref_segment_a, ref_segment_b});

    ref_table->append_chunk(segments);

    return ref_table;
  }

  void scan_for_null_values(const std::shared_ptr<AbstractOperator> in,
                            const std::map<PredicateCondition, std::vector<AllTypeVariant>>& tests) {
    for (const auto& test : tests) {
      const auto predicate_condition = test.first;
      const auto& expected = test.second;  // {12, 123}

      const auto column = get_column_expression(in, ColumnID{1});

      auto scan = create_table_scan(in, ColumnID{1}, predicate_condition, NULL_VALUE);
      scan->execute();

      ASSERT_COLUMN_EQ(scan->get_output(), ColumnID{0}, expected);
    }
  }

  void ASSERT_COLUMN_EQ(std::shared_ptr<const Table> table, const ColumnID& column_id,
                        std::vector<AllTypeVariant> expected) {
    const auto chunk_count = table->chunk_count();
    for (auto chunk_id = ChunkID{0}; chunk_id < chunk_count; ++chunk_id) {
      const auto chunk = table->get_chunk(chunk_id);

      const auto chunk_size = chunk->size();
      for (auto chunk_offset = ChunkOffset{0}; chunk_offset < chunk_size; ++chunk_offset) {
        const auto& segment = *chunk->get_segment(column_id);

        const auto found_value = segment[chunk_offset];
        const auto comparator = [found_value](const AllTypeVariant expected_value) {
          // returns equivalency, not equality to simulate std::multiset.
          // multiset cannot be used because it triggers a compiler / lib bug when built in CI
          return !(found_value < expected_value) && !(expected_value < found_value);
        };

        auto search = std::find_if(expected.begin(), expected.end(), comparator);

        ASSERT_TRUE(search != expected.end()) << found_value << " not found";
        expected.erase(search);
      }
    }

    ASSERT_EQ(expected.size(), 0u);
  }

  void scan_and_check_sorted_by(const std::shared_ptr<TableWrapper> table_wrapper) const {
    const auto scan_sorted = create_table_scan(table_wrapper, ColumnID{0}, PredicateCondition::GreaterThanEquals, 1234);
    scan_sorted->execute();
    const auto& result_table_sorted = scan_sorted->get_output();

    const auto chunk_count = result_table_sorted->chunk_count();
    for (auto chunk_id = ChunkID{0}; chunk_id < chunk_count; ++chunk_id) {
      const auto& actual_sorted_by = result_table_sorted->get_chunk(chunk_id)->individually_sorted_by();
      ASSERT_TRUE(!actual_sorted_by.empty());
      const auto expected_sorted_by =
          std::vector<SortColumnDefinition>{SortColumnDefinition(ColumnID{0}, SortMode::Ascending)};
      EXPECT_EQ(actual_sorted_by, expected_sorted_by);
    }
  }

 protected:
  EncodingType _encoding_type;
  std::shared_ptr<TableWrapper> _int_int_compressed;
  std::shared_ptr<TableWrapper> _int_int_partly_compressed;
};

INSTANTIATE_TEST_SUITE_P(EncodingTypes, OperatorsTableScanTest,
                         ::testing::Values(EncodingType::Unencoded, EncodingType::Dictionary, EncodingType::RunLength,
                                           EncodingType::FrameOfReference),
                         enum_formatter<EncodingType>);

TEST_P(OperatorsTableScanTest, DoubleScan) {
  const auto expected_result = load_table("resources/test_data/tbl/int_float_filtered.tbl", ChunkOffset{2});

  auto scan_1 = create_table_scan(get_int_float_op(), ColumnID{0}, PredicateCondition::GreaterThanEquals, 1234);
  scan_1->execute();

  auto scan_2 = create_table_scan(scan_1, ColumnID{1}, PredicateCondition::LessThan, 457.9);
  scan_2->execute();

  EXPECT_TABLE_EQ_UNORDERED(scan_2->get_output(), expected_result);
}

TEST_P(OperatorsTableScanTest, EmptyResultScan) {
  auto scan_1 = create_table_scan(get_int_float_op(), ColumnID{0}, PredicateCondition::GreaterThan, 90000);
  scan_1->execute();

  for (auto i = ChunkID{0}; i < scan_1->get_output()->chunk_count(); i++) {
    EXPECT_EQ(scan_1->get_output()->get_chunk(i)->column_count(), 2u);
  }
}

TEST_P(OperatorsTableScanTest, SingleScan) {
  const auto expected_result = load_table("resources/test_data/tbl/int_float_filtered2.tbl", ChunkOffset{1});

  auto scan = create_table_scan(get_int_float_op(), ColumnID{0}, PredicateCondition::GreaterThanEquals, 1234);
  scan->execute();

  EXPECT_TABLE_EQ_UNORDERED(scan->get_output(), expected_result);
}

TEST_P(OperatorsTableScanTest, SingleScanWithSortedSegmentEquals) {
  const auto expected_result = load_table("resources/test_data/tbl/int_sorted_filtered.tbl", ChunkOffset{1});

  auto scan = create_table_scan(get_int_sorted_op(), ColumnID{0}, PredicateCondition::Equals, 2);
  scan->execute();

  EXPECT_TABLE_EQ_UNORDERED(scan->get_output(), expected_result);
}

TEST_P(OperatorsTableScanTest, SingleScanWithSortedSegmentEqualsAllElementsEqualNull) {
  const auto expected_result = load_table("resources/test_data/tbl/int_empty_nullable.tbl", ChunkOffset{1});

  auto scan = create_table_scan(get_int_only_null_op(), ColumnID{0}, PredicateCondition::Equals, 2);
  scan->execute();

  EXPECT_TABLE_EQ_UNORDERED(scan->get_output(), expected_result);
}

TEST_P(OperatorsTableScanTest, SingleScanWithSortedSegmentEqualsAllPredicateValueLarger) {
  const auto expected_result = load_table("resources/test_data/tbl/int_empty.tbl", ChunkOffset{1});

  auto scan = create_table_scan(get_int_sorted_op(), ColumnID{0}, PredicateCondition::Equals, 6);
  scan->execute();

  EXPECT_TABLE_EQ_UNORDERED(scan->get_output(), expected_result);
}

TEST_P(OperatorsTableScanTest, SingleScanWithSortedSegmentNotEquals) {
  const auto expected_result = load_table("resources/test_data/tbl/int_sorted_filtered2.tbl", ChunkOffset{1});

  auto scan = create_table_scan(get_int_sorted_op(), ColumnID{0}, PredicateCondition::NotEquals, 2);
  scan->execute();

  EXPECT_TABLE_EQ_UNORDERED(scan->get_output(), expected_result);
}

TEST_P(OperatorsTableScanTest, SingleScanWithSubquery) {
  const auto expected_result = load_table("resources/test_data/tbl/int_float_filtered2.tbl", ChunkOffset{1});

  const auto subquery_pqp = std::make_shared<Projection>(std::make_shared<TableWrapper>(Projection::dummy_table()),
                                                         expression_vector(value_(1234)));
  execute_all({subquery_pqp->mutable_left_input(), subquery_pqp});

  const auto scan = std::make_shared<TableScan>(
      get_int_float_op(), greater_than_equals_(pqp_column_(ColumnID{0}, DataType::Int, false, "a"),
                                               pqp_subquery_(subquery_pqp, DataType::Int, false)));
  EXPECT_TRUE(dynamic_cast<ColumnVsValueTableScanImpl*>(scan->create_impl().get()));
  scan->execute();
  EXPECT_TABLE_EQ_UNORDERED(scan->get_output(), expected_result);
}

TEST_P(OperatorsTableScanTest, BetweenScanWithSubquery) {
  auto expected_result = load_table("resources/test_data/tbl/int_float_filtered2.tbl", ChunkOffset{1});

  const auto subquery_pqp = std::make_shared<Projection>(std::make_shared<TableWrapper>(Projection::dummy_table()),
                                                         expression_vector(value_(1234)));
  execute_all({subquery_pqp->mutable_left_input(), subquery_pqp});

  const auto scan = std::make_shared<TableScan>(
      get_int_float_op(), between_inclusive_(pqp_column_(ColumnID{0}, DataType::Int, false, "a"),
                                             pqp_subquery_(subquery_pqp, DataType::Int, false), value_(12345)));
  EXPECT_TRUE(dynamic_cast<ColumnBetweenTableScanImpl*>(scan->create_impl().get()));
  scan->execute();
  EXPECT_TABLE_EQ_UNORDERED(scan->get_output(), expected_result);
}

TEST_P(OperatorsTableScanTest, SingleScanWithEmptySubquery) {
  const auto expected_result = Table::create_dummy_table(get_int_float_op()->table->column_definitions());
  const auto dummy_table = Table::create_dummy_table({{"dummy", DataType::Int, false}});

  const auto subquery_pqp = std::make_shared<TableWrapper>(dummy_table);
  subquery_pqp->execute();

  const auto scan = std::make_shared<TableScan>(
      get_int_float_op(), greater_than_equals_(pqp_column_(ColumnID{0}, DataType::Int, false, "a"),
                                               pqp_subquery_(subquery_pqp, DataType::Int, false)));
  EXPECT_TRUE(dynamic_cast<ExpressionEvaluatorTableScanImpl*>(scan->create_impl().get()));
  scan->execute();
  EXPECT_TABLE_EQ_UNORDERED(scan->get_output(), expected_result);
}

TEST_P(OperatorsTableScanTest, SingleScanWithInvalidSubquery) {
  // Uncorrelated subqueries must return one or zero rows.
  const auto subquery_pqp = std::make_shared<Projection>(get_int_string_op(), expression_vector(value_(1234)));
  subquery_pqp->execute();

  const auto scan = std::make_shared<TableScan>(
      get_int_float_op(), greater_than_equals_(pqp_column_(ColumnID{0}, DataType::Int, false, "a"),
                                               pqp_subquery_(subquery_pqp, DataType::Int, false)));

  EXPECT_THROW(scan->create_impl(), std::logic_error);
}

TEST_P(OperatorsTableScanTest, BetweenScanWithInvalidSubquery) {
  // Uncorrelated subqueries must return one or zero rows.
  const auto subquery_pqp = std::make_shared<Projection>(get_int_string_op(), expression_vector(value_(1234)));
  subquery_pqp->execute();

  const auto scan = std::make_shared<TableScan>(
      get_int_float_op(), between_inclusive_(pqp_column_(ColumnID{0}, DataType::Int, false, "a"),
                                             pqp_subquery_(subquery_pqp, DataType::Int, false), value_(12345)));

  EXPECT_THROW(scan->create_impl(), std::logic_error);
}

TEST_P(OperatorsTableScanTest, ScanOnCompressedSegments) {
  // We do not need to check for a non existing value, because that happens automatically when we scan the second chunk.

  auto tests = std::map<PredicateCondition, std::vector<AllTypeVariant>>{};
  tests[PredicateCondition::Equals] = {106, 106};
  tests[PredicateCondition::NotEquals] = {100, 102, 104, 108, 110, 112, 100, 102, 104, 108, 110, 112};
  tests[PredicateCondition::LessThan] = {100, 102, 104, 100, 102, 104};
  tests[PredicateCondition::LessThanEquals] = {100, 102, 104, 106, 100, 102, 104, 106};
  tests[PredicateCondition::GreaterThan] = {108, 110, 112, 108, 110, 112};
  tests[PredicateCondition::GreaterThanEquals] = {106, 108, 110, 112, 106, 108, 110, 112};
  tests[PredicateCondition::IsNull] = {};
  tests[PredicateCondition::IsNotNull] = {100, 102, 104, 106, 108, 110, 112, 100, 102, 104, 106, 108, 110, 112};

  for (const auto& test : tests) {
    auto scan_int = create_table_scan(_int_int_compressed, ColumnID{0}, test.first, 6);
    auto scan_int_partly = create_table_scan(_int_int_partly_compressed, ColumnID{0}, test.first, 6);

    scan_int->execute();

    ASSERT_COLUMN_EQ(scan_int->get_output(), ColumnID{1}, test.second);

    scan_int_partly->execute();
    ASSERT_COLUMN_EQ(scan_int_partly->get_output(), ColumnID{1}, test.second);
  }
}

TEST_P(OperatorsTableScanTest, ScanOnReferencedCompressedSegments) {
  // we do not need to check for a non existing value, because that happens automatically when we scan the second chunk

  std::map<PredicateCondition, std::vector<AllTypeVariant>> tests;
  tests[PredicateCondition::Equals] = {104, 104};
  tests[PredicateCondition::NotEquals] = {100, 102, 106, 100, 102, 106};
  tests[PredicateCondition::LessThan] = {100, 102, 100, 102};
  tests[PredicateCondition::LessThanEquals] = {100, 102, 104, 100, 102, 104};
  tests[PredicateCondition::GreaterThan] = {106, 106};
  tests[PredicateCondition::GreaterThanEquals] = {104, 106, 104, 106};
  tests[PredicateCondition::IsNull] = {};
  tests[PredicateCondition::IsNotNull] = {100, 102, 104, 106, 100, 102, 104, 106};

  for (const auto& [predicate_condition, expected] : tests) {
    // the first scans will match all values since the predicate condition evaluates to true for all
    auto scan1 = create_table_scan(_int_int_compressed, ColumnID{1}, PredicateCondition::LessThan, 108);
    scan1->execute();

    auto scan_partly1 = create_table_scan(_int_int_partly_compressed, ColumnID{1}, PredicateCondition::LessThan, 108);
    scan_partly1->execute();

    auto scan2 = create_table_scan(scan1, ColumnID{0}, predicate_condition, 4);
    auto scan_partly2 = create_table_scan(scan_partly1, ColumnID{0}, predicate_condition, 4);

    scan2->execute();
    scan_partly2->execute();

    ASSERT_COLUMN_EQ(scan2->get_output(), ColumnID{1}, expected);
    ASSERT_COLUMN_EQ(scan_partly2->get_output(), ColumnID{1}, expected);
  }
}

TEST_P(OperatorsTableScanTest, ScanWeirdPosList) {
  std::map<PredicateCondition, std::vector<AllTypeVariant>> tests;
  tests[PredicateCondition::Equals] = {110, 110};
  tests[PredicateCondition::NotEquals] = {100, 102, 106, 108, 112};
  tests[PredicateCondition::LessThan] = {100, 102, 106, 108};
  tests[PredicateCondition::LessThanEquals] = {100, 102, 106, 108, 110, 110};
  tests[PredicateCondition::GreaterThan] = {112};
  tests[PredicateCondition::GreaterThanEquals] = {110, 110, 112};
  tests[PredicateCondition::IsNull] = {};
  tests[PredicateCondition::IsNotNull] = {100, 102, 106, 108, 110, 110, 112};

  auto table_wrapper = get_table_op_filtered();

  for (const auto& test : tests) {
    auto scan_partly = create_table_scan(table_wrapper, ColumnID{0}, test.first, 10);
    scan_partly->execute();

    ASSERT_COLUMN_EQ(scan_partly->get_output(), ColumnID{1}, test.second);
  }
}

TEST_P(OperatorsTableScanTest, ScanOnCompressedSegmentsValueGreaterThanMaxDictionaryValue) {
  const auto all_rows =
      std::vector<AllTypeVariant>{100, 102, 104, 106, 108, 110, 112, 100, 102, 104, 106, 108, 110, 112};
  const auto no_rows = std::vector<AllTypeVariant>{};

  std::map<PredicateCondition, std::vector<AllTypeVariant>> tests;
  tests[PredicateCondition::Equals] = no_rows;
  tests[PredicateCondition::NotEquals] = all_rows;
  tests[PredicateCondition::LessThan] = all_rows;
  tests[PredicateCondition::LessThanEquals] = all_rows;
  tests[PredicateCondition::GreaterThan] = no_rows;
  tests[PredicateCondition::GreaterThanEquals] = no_rows;

  for (const auto& test : tests) {
    auto scan = create_table_scan(_int_int_compressed, ColumnID{0}, test.first, 30);
    scan->execute();

    auto scan_partly = create_table_scan(_int_int_partly_compressed, ColumnID{0}, test.first, 30);
    scan_partly->execute();

    ASSERT_COLUMN_EQ(scan->get_output(), ColumnID{1}, test.second);
    ASSERT_COLUMN_EQ(scan_partly->get_output(), ColumnID{1}, test.second);
  }
}

TEST_P(OperatorsTableScanTest, ScanOnCompressedSegmentsValueLessThanMinDictionaryValue) {
  const auto all_rows =
      std::vector<AllTypeVariant>{100, 102, 104, 106, 108, 110, 112, 100, 102, 104, 106, 108, 110, 112};
  const auto no_rows = std::vector<AllTypeVariant>{};

  std::map<PredicateCondition, std::vector<AllTypeVariant>> tests;
  tests[PredicateCondition::Equals] = no_rows;
  tests[PredicateCondition::NotEquals] = all_rows;
  tests[PredicateCondition::LessThan] = no_rows;
  tests[PredicateCondition::LessThanEquals] = no_rows;
  tests[PredicateCondition::GreaterThan] = all_rows;
  tests[PredicateCondition::GreaterThanEquals] = all_rows;

  for (const auto& test : tests) {
    auto scan = create_table_scan(_int_int_compressed, ColumnID{0} /* "a" */, test.first, -10);
    scan->execute();

    auto scan_partly = create_table_scan(_int_int_partly_compressed, ColumnID{0} /* "a" */, test.first, -10);
    scan_partly->execute();

    ASSERT_COLUMN_EQ(scan->get_output(), ColumnID{1}, test.second);
    ASSERT_COLUMN_EQ(scan_partly->get_output(), ColumnID{1}, test.second);
  }
}

TEST_P(OperatorsTableScanTest, ScanOnIntValueSegmentWithFloatColumnWithNullValues) {
  auto table = load_table("resources/test_data/tbl/int_int_w_null_8_rows.tbl", ChunkOffset{4});

  auto table_wrapper = std::make_shared<TableWrapper>(std::move(table));
  table_wrapper->execute();

  const auto predicate = greater_than_(get_column_expression(table_wrapper, ColumnID{0}),
                                       get_column_expression(table_wrapper, ColumnID{1}));
  auto scan = std::make_shared<TableScan>(table_wrapper, predicate);
  scan->execute();

  const auto expected = std::vector<AllTypeVariant>{12345, 1234, 12345, 1234};
  ASSERT_COLUMN_EQ(scan->get_output(), ColumnID{0}, expected);
}

TEST_P(OperatorsTableScanTest, ScanOnReferencedIntValueSegmentWithFloatColumnWithNullValues) {
  auto table = load_table("resources/test_data/tbl/int_int_w_null_8_rows.tbl", ChunkOffset{4});

  auto table_wrapper = std::make_shared<TableWrapper>(to_simple_reference_table(table));
  table_wrapper->execute();

  const auto predicate = greater_than_(get_column_expression(table_wrapper, ColumnID{0}),
                                       get_column_expression(table_wrapper, ColumnID{1}));
  auto scan = std::make_shared<TableScan>(table_wrapper, predicate);
  scan->execute();

  const auto expected = std::vector<AllTypeVariant>{12345, 1234, 12345, 1234};
  ASSERT_COLUMN_EQ(scan->get_output(), ColumnID{0}, expected);
}

TEST_P(OperatorsTableScanTest, ScanOnIntCompressedSegmentsWithFloatColumnWithNullValues) {
  auto table = load_table("resources/test_data/tbl/int_int_w_null_8_rows.tbl", ChunkOffset{4});
  ChunkEncoder::encode_all_chunks(table, SegmentEncodingSpec{_encoding_type});

  auto table_wrapper = std::make_shared<TableWrapper>(std::move(table));
  table_wrapper->execute();

  const auto predicate = greater_than_(get_column_expression(table_wrapper, ColumnID{0}),
                                       get_column_expression(table_wrapper, ColumnID{1}));
  auto scan = std::make_shared<TableScan>(table_wrapper, predicate);
  scan->execute();

  const auto expected = std::vector<AllTypeVariant>{12345, 1234, 12345, 1234};
  ASSERT_COLUMN_EQ(scan->get_output(), ColumnID{0}, expected);
}

TEST_P(OperatorsTableScanTest, ScanOnReferencedIntCompressedSegmentsWithFloatColumnWithNullValues) {
  auto table = load_table("resources/test_data/tbl/int_int_w_null_8_rows.tbl", ChunkOffset{4});
  ChunkEncoder::encode_all_chunks(table, SegmentEncodingSpec{_encoding_type});

  auto table_wrapper = std::make_shared<TableWrapper>(to_simple_reference_table(table));
  table_wrapper->execute();

  const auto predicate = greater_than_(get_column_expression(table_wrapper, ColumnID{0}),
                                       get_column_expression(table_wrapper, ColumnID{1}));
  auto scan = std::make_shared<TableScan>(table_wrapper, predicate);
  scan->execute();

  const auto expected = std::vector<AllTypeVariant>{12345, 1234, 12345, 1234};
  ASSERT_COLUMN_EQ(scan->get_output(), ColumnID{0}, expected);
}

TEST_P(OperatorsTableScanTest, ScanOnCompressedSegmentsAroundBounds) {
  // scanning for a value that is around the dictionary's bounds

  std::map<PredicateCondition, std::vector<AllTypeVariant>> tests;
  tests[PredicateCondition::Equals] = {100, 100};
  tests[PredicateCondition::LessThan] = {};
  tests[PredicateCondition::LessThanEquals] = {100, 100};
  tests[PredicateCondition::GreaterThan] = {102, 104, 106, 108, 110, 112, 102, 104, 106, 108, 110, 112};
  tests[PredicateCondition::GreaterThanEquals] = {100, 102, 104, 106, 108, 110, 112, 100, 102, 104, 106, 108, 110, 112};
  tests[PredicateCondition::NotEquals] = {102, 104, 106, 108, 110, 112, 102, 104, 106, 108, 110, 112};
  tests[PredicateCondition::IsNull] = {};
  tests[PredicateCondition::IsNotNull] = {100, 102, 104, 106, 108, 110, 112, 100, 102, 104, 106, 108, 110, 112};

  for (const auto& test : tests) {
    auto scan = create_table_scan(_int_int_compressed, ColumnID{0}, test.first, 0);
    scan->execute();

    auto scan_partly = create_table_scan(_int_int_partly_compressed, ColumnID{0}, test.first, 0);
    scan_partly->execute();

    ASSERT_COLUMN_EQ(scan->get_output(), ColumnID{1}, test.second);
    ASSERT_COLUMN_EQ(scan_partly->get_output(), ColumnID{1}, test.second);
  }
}

TEST_P(OperatorsTableScanTest, ScanWithEmptyInput) {
  auto scan_1 = std::make_shared<TableScan>(
      get_int_float_op(), greater_than_(get_column_expression(get_int_float_op(), ColumnID{0}), 12345));
  scan_1->execute();
  EXPECT_EQ(scan_1->get_output()->row_count(), static_cast<size_t>(0));

  // scan_1 produced an empty result
  auto scan_2 = std::make_shared<TableScan>(scan_1, equals_(get_column_expression(scan_1, ColumnID{1}), 456.7));
  scan_2->execute();

  EXPECT_EQ(scan_2->get_output()->row_count(), static_cast<size_t>(0));
}

TEST_P(OperatorsTableScanTest, ScanOnWideDictionarySegment) {
  // 2**8 + 1 values require a data type of 16bit.
  const auto table_wrapper_dict_16 = get_table_op_with_n_dict_entries((1 << 8) + 1);
  auto scan_1 = std::make_shared<TableScan>(
      table_wrapper_dict_16, greater_than_(get_column_expression(table_wrapper_dict_16, ColumnID{0}), 200));
  scan_1->execute();

  EXPECT_EQ(scan_1->get_output()->row_count(), static_cast<size_t>(57));

  // 2**16 + 1 values require a data type of 32bit.
  const auto table_wrapper_dict_32 = get_table_op_with_n_dict_entries((1 << 16) + 1);
  auto scan_2 = std::make_shared<TableScan>(
      table_wrapper_dict_32, greater_than_(get_column_expression(table_wrapper_dict_32, ColumnID{0}), 65500));
  scan_2->execute();

  EXPECT_EQ(scan_2->get_output()->row_count(), static_cast<size_t>(37));
}

TEST_P(OperatorsTableScanTest, OperatorName) {
  auto scan_1 = std::make_shared<TableScan>(
      get_int_float_op(), greater_than_(get_column_expression(get_int_float_op(), ColumnID{0}), 12345));

  EXPECT_EQ(scan_1->name(), "TableScan");
}

TEST_P(OperatorsTableScanTest, ScanForNullValuesOnValueSegment) {
  auto table_wrapper =
      std::make_shared<TableWrapper>(load_table("resources/test_data/tbl/int_int_w_null_8_rows.tbl", ChunkOffset{4}));
  table_wrapper->never_clear_output();
  table_wrapper->execute();

  const auto tests = std::map<PredicateCondition, std::vector<AllTypeVariant>>{
      {PredicateCondition::IsNull, {12, 123}},
      {PredicateCondition::IsNotNull, {12345, NULL_VALUE, 1234, 12345, 12, 1234}}};

  scan_for_null_values(table_wrapper, tests);
}

TEST_P(OperatorsTableScanTest, ScanForNullValuesOnCompressedSegments) {
  auto table = load_table("resources/test_data/tbl/int_int_w_null_8_rows.tbl", ChunkOffset{4});
  ChunkEncoder::encode_all_chunks(table, SegmentEncodingSpec{_encoding_type});

  auto table_wrapper = std::make_shared<TableWrapper>(table);
  table_wrapper->never_clear_output();
  table_wrapper->execute();

  const auto tests = std::map<PredicateCondition, std::vector<AllTypeVariant>>{
      {PredicateCondition::IsNull, {12, 123}},
      {PredicateCondition::IsNotNull, {12345, NULL_VALUE, 1234, 12345, 12, 1234}}};

  scan_for_null_values(table_wrapper, tests);
}

TEST_P(OperatorsTableScanTest, ScanForNullValuesOnCompressedSortedSegments) {
  const auto table = load_table("resources/test_data/tbl/int_null_sorted_asc_2.tbl", ChunkOffset{4});
  table->get_chunk(ChunkID{0})->set_individually_sorted_by(SortColumnDefinition(ColumnID{1}, SortMode::Ascending));
  ChunkEncoder::encode_all_chunks(table, SegmentEncodingSpec{_encoding_type});

  const auto table_wrapper = std::make_shared<TableWrapper>(table);
  table_wrapper->never_clear_output();
  table_wrapper->execute();

  const auto tests = std::map<PredicateCondition, std::vector<AllTypeVariant>>{{PredicateCondition::IsNull, {1, 2}},
                                                                               {PredicateCondition::IsNotNull, {3, 4}}};

  scan_for_null_values(table_wrapper, tests);
}

TEST_P(OperatorsTableScanTest, ScanForNullValuesOnCompressedDescendingSortedSegments) {
  const auto table = load_table("resources/test_data/tbl/int_null_sorted_desc_2.tbl", ChunkOffset{4});
  table->get_chunk(ChunkID{0})->set_individually_sorted_by(SortColumnDefinition(ColumnID{1}, SortMode::Descending));
  ChunkEncoder::encode_all_chunks(table, SegmentEncodingSpec{_encoding_type});

  const auto table_wrapper = std::make_shared<TableWrapper>(table);
  table_wrapper->never_clear_output();
  table_wrapper->execute();

  const auto tests = std::map<PredicateCondition, std::vector<AllTypeVariant>>{{PredicateCondition::IsNull, {1, 2}},
                                                                               {PredicateCondition::IsNotNull, {3, 4}}};

  scan_for_null_values(table_wrapper, tests);
}

TEST_P(OperatorsTableScanTest, ScanForNullValuesOnValueSegmentWithoutNulls) {
  auto table = load_table("resources/test_data/tbl/int_float.tbl", ChunkOffset{4});

  auto table_wrapper = std::make_shared<TableWrapper>(table);
  table_wrapper->never_clear_output();
  table_wrapper->execute();

  const auto tests = std::map<PredicateCondition, std::vector<AllTypeVariant>>{
      {PredicateCondition::IsNull, {}}, {PredicateCondition::IsNotNull, {12345, 123, 1234}}};

  scan_for_null_values(table_wrapper, tests);
}

TEST_P(OperatorsTableScanTest, ScanForNullValuesOnReferencedValueSegmentWithoutNulls) {
  auto table = load_table("resources/test_data/tbl/int_float.tbl", ChunkOffset{4});

  auto table_wrapper = std::make_shared<TableWrapper>(to_simple_reference_table(table));
  table_wrapper->never_clear_output();
  table_wrapper->execute();

  const auto tests = std::map<PredicateCondition, std::vector<AllTypeVariant>>{
      {PredicateCondition::IsNull, {}}, {PredicateCondition::IsNotNull, {12345, 123, 1234}}};

  scan_for_null_values(table_wrapper, tests);
}

TEST_P(OperatorsTableScanTest, ScanForNullValuesOnReferencedValueSegment) {
  auto table = load_table("resources/test_data/tbl/int_int_w_null_8_rows.tbl", ChunkOffset{4});

  auto table_wrapper = std::make_shared<TableWrapper>(to_simple_reference_table(table));
  table_wrapper->never_clear_output();
  table_wrapper->execute();

  const auto tests = std::map<PredicateCondition, std::vector<AllTypeVariant>>{
      {PredicateCondition::IsNull, {12, 123}},
      {PredicateCondition::IsNotNull, {12345, NULL_VALUE, 1234, 12345, 12, 1234}}};

  scan_for_null_values(table_wrapper, tests);
}

TEST_P(OperatorsTableScanTest, ScanForNullValuesOnReferencedCompressedSegments) {
  auto table = load_table("resources/test_data/tbl/int_int_w_null_8_rows.tbl", ChunkOffset{4});
  ChunkEncoder::encode_all_chunks(table, SegmentEncodingSpec{_encoding_type});

  auto table_wrapper = std::make_shared<TableWrapper>(to_simple_reference_table(table));
  table_wrapper->never_clear_output();
  table_wrapper->execute();

  const auto tests = std::map<PredicateCondition, std::vector<AllTypeVariant>>{
      {PredicateCondition::IsNull, {12, 123}},
      {PredicateCondition::IsNotNull, {12345, NULL_VALUE, 1234, 12345, 12, 1234}}};

  scan_for_null_values(table_wrapper, tests);
}

TEST_P(OperatorsTableScanTest, ScanForNullValuesWithNullRowIDOnReferencedValueSegment) {
  auto table = create_referencing_table_w_null_row_id(false);

  auto table_wrapper = std::make_shared<TableWrapper>(table);
  table_wrapper->never_clear_output();
  table_wrapper->execute();

  const auto tests = std::map<PredicateCondition, std::vector<AllTypeVariant>>{
      {PredicateCondition::IsNull, {123, 1234}}, {PredicateCondition::IsNotNull, {12345, NULL_VALUE}}};

  scan_for_null_values(table_wrapper, tests);
}

TEST_P(OperatorsTableScanTest, ScanForNullValuesWithNullRowIDOnReferencedCompressedSegments) {
  auto table = create_referencing_table_w_null_row_id(true);

  auto table_wrapper = std::make_shared<TableWrapper>(table);
  table_wrapper->never_clear_output();
  table_wrapper->execute();

  const auto tests = std::map<PredicateCondition, std::vector<AllTypeVariant>>{
      {PredicateCondition::IsNull, {123, 1234}}, {PredicateCondition::IsNotNull, {12345, NULL_VALUE}}};

  scan_for_null_values(table_wrapper, tests);
}

TEST_P(OperatorsTableScanTest, ComparisonOfIntColumnAndNullValue) {
  const auto predicate_conditions = std::vector<PredicateCondition>(
      {PredicateCondition::Equals, PredicateCondition::NotEquals, PredicateCondition::LessThan,
       PredicateCondition::LessThanEquals, PredicateCondition::GreaterThan, PredicateCondition::GreaterThanEquals});

  for (auto predicate_condition : predicate_conditions) {
    auto scan = create_table_scan(get_int_float_with_null_op(), ColumnID{0}, predicate_condition, NullValue{});
    scan->execute();

    EXPECT_EQ(scan->get_output()->row_count(), 0u);

    for (auto chunk_id = ChunkID{0}; chunk_id < scan->get_output()->chunk_count(); chunk_id++) {
      EXPECT_EQ(scan->get_output()->get_chunk(chunk_id)->column_count(), 2u);
    }
  }
}

TEST_P(OperatorsTableScanTest, ComparisonOfStringColumnAndNullValue) {
  const auto predicate_conditions = std::vector<PredicateCondition>(
      {PredicateCondition::Equals, PredicateCondition::NotEquals, PredicateCondition::LessThan,
       PredicateCondition::LessThanEquals, PredicateCondition::GreaterThan, PredicateCondition::GreaterThanEquals});

  for (auto predicate_condition : predicate_conditions) {
    auto scan = create_table_scan(get_int_string_op(), ColumnID{1}, predicate_condition, NullValue{});
    scan->execute();

    EXPECT_EQ(scan->get_output()->row_count(), 0u);
    for (auto chunk_id = ChunkID{0}; chunk_id < scan->get_output()->chunk_count(); chunk_id++) {
      EXPECT_EQ(scan->get_output()->get_chunk(chunk_id)->column_count(), 2u);
    }
  }
}

TEST_P(OperatorsTableScanTest, MatchesAllExcludesNulls) {
  // Scan implementations will potentially optimize the scan if they can detect that all values in the column match the
  // predicate.
  // Test that if the predicate matches all values in the Chunk, rows with NULL values in the scanned column are
  // nevertheless excluded from the result. E.g. [NULL, 1, 2, NULL] < 5 = [1, 2]

  /**
   * Binary Predicates
   */
  auto predicates = std::vector<std::tuple<ColumnID, PredicateCondition, AllTypeVariant, std::vector<AllTypeVariant>>>{
      {ColumnID{0}, PredicateCondition::Equals, 1234, {1234}},  // Matches all in second chunk
      {ColumnID{0}, PredicateCondition::NotEquals, 100, {12345, 123, 1234}},
      {ColumnID{0}, PredicateCondition::LessThan, 15'000, {12345, 123, 1234}},
      {ColumnID{0}, PredicateCondition::LessThanEquals, 15'000, {12345, 123, 1234}},
      {ColumnID{0}, PredicateCondition::GreaterThan, 0, {12345, 123, 1234}},
      {ColumnID{0}, PredicateCondition::GreaterThanEquals, 0, {12345, 123, 1234}}};

  const auto table = get_int_float_with_null_op();
  for (const auto& [column_id, predicate_condition, value, expected_values] : predicates) {
    const auto scan = create_table_scan(table, column_id, predicate_condition, value);
    scan->execute();
    ASSERT_COLUMN_EQ(scan->get_output(), column_id, expected_values);
  }

  /**
   * BETWEEN
   */
  const auto between_scan = create_table_scan(table, ColumnID{0}, PredicateCondition::BetweenInclusive, 0, 15'000);
  between_scan->execute();
  ASSERT_COLUMN_EQ(between_scan->get_output(), ColumnID{0}, {12345, 123, 1234});

  /**
   * IS NULL
   */
  // Second Chunk of Column 1 has only NULL values
  const auto table_with_null_chunk = load_and_encode_table("resources/test_data/tbl/int_int_int_null.tbl");
  const auto is_null_scan =
      create_table_scan(table_with_null_chunk, ColumnID{1}, PredicateCondition::IsNull, NullValue{});
  is_null_scan->execute();
  ASSERT_COLUMN_EQ(is_null_scan->get_output(), ColumnID{0}, {11, 9});
  ASSERT_COLUMN_EQ(is_null_scan->get_output(), ColumnID{1}, {NullValue{}, NullValue{}});

  /**
   * IS NOT NULL
   */
  // First Chunk of Column 0 has no NULL values
  const auto is_not_null_scan = create_table_scan(table, ColumnID{0}, PredicateCondition::IsNotNull, NullValue{});
  is_not_null_scan->execute();
  ASSERT_COLUMN_EQ(is_not_null_scan->get_output(), ColumnID{0}, {12345, 123, 1234});
}

TEST_P(OperatorsTableScanTest, ScanWithExcludedFirstChunk) {
  const auto expected = std::vector<AllTypeVariant>{100, 102, 104, 106, 108, 110, 112, 102, 104};

  auto scan = std::make_shared<TableScan>(
      _int_int_partly_compressed,
      greater_than_equals_(get_column_expression(_int_int_partly_compressed, ColumnID{0}), 0));
  scan->excluded_chunk_ids = std::make_shared<std::vector<ChunkID>>(std::initializer_list<ChunkID>{ChunkID{0}});
  scan->execute();

  ASSERT_COLUMN_EQ(scan->get_output(), ColumnID{1}, expected);
}

TEST_P(OperatorsTableScanTest, BinaryScanOnNullable) {
  auto predicates = std::vector<std::tuple<ColumnID, PredicateCondition, AllTypeVariant, std::vector<AllTypeVariant>>>{
      {ColumnID{0}, PredicateCondition::Equals, 1234, {1234}},
      {ColumnID{0}, PredicateCondition::NotEquals, 123, {12345, 1234}},
      {ColumnID{0}, PredicateCondition::GreaterThan, 123, {12345, 1234}},
      {ColumnID{0}, PredicateCondition::GreaterThanEquals, 124, {12345, 1234}},
      {ColumnID{0}, PredicateCondition::LessThan, 1235, {123, 1234}},
      {ColumnID{0}, PredicateCondition::LessThanEquals, 1234, {123, 1234}}};

  const auto table = get_int_float_with_null_op(Chunk::DEFAULT_SIZE);
  for (const auto& [column_id, predicate_condition, value, expected_values] : predicates) {
    const auto scan = create_table_scan(table, column_id, predicate_condition, value);
    scan->execute();
    ASSERT_COLUMN_EQ(scan->get_output(), column_id, expected_values);
  }
}

TEST_P(OperatorsTableScanTest, BinaryScanWithFloatValueOnIntColumn) {
  auto predicates = std::vector<std::tuple<ColumnID, PredicateCondition, AllTypeVariant, std::vector<AllTypeVariant>>>{
      {ColumnID{0}, PredicateCondition::Equals, 1234.0f, {1234}},
      {ColumnID{0}, PredicateCondition::Equals, 1234.1f, {}},
      {ColumnID{0}, PredicateCondition::NotEquals, 123.0f, {12345, 1234}},
      {ColumnID{0}, PredicateCondition::NotEquals, 123.1f, {12345, 123, 1234}},
      {ColumnID{0}, PredicateCondition::GreaterThan, 1234.1f, {12345}},
      {ColumnID{0}, PredicateCondition::GreaterThanEquals, 1234.1f, {12345}},
      {ColumnID{0}, PredicateCondition::LessThan, 1234.0, {123}},
      {ColumnID{0}, PredicateCondition::LessThan, 1234.1, {123, 1234}},
      {ColumnID{0}, PredicateCondition::LessThanEquals, 1234.0, {123, 1234}},
      {ColumnID{0}, PredicateCondition::LessThanEquals, 1234.1, {123, 1234}}};

  const auto table = get_int_float_op();
  for (const auto& [column_id, predicate_condition, value, expected_values] : predicates) {
    const auto scan = create_table_scan(table, column_id, predicate_condition, value);
    scan->execute();
    ASSERT_COLUMN_EQ(scan->get_output(), column_id, expected_values);
  }
}

TEST_P(OperatorsTableScanTest, SetParameters) {
  const auto parameters = std::unordered_map<ParameterID, AllTypeVariant>{{ParameterID{3}, AllTypeVariant{5}},
                                                                          {ParameterID{2}, AllTypeVariant{6}}};

  const auto column = get_column_expression(_int_int_compressed, ColumnID{0});

  // Hint: To set parameters, operators are not allowed to have executed.
  //       Therefore, we deep copy the TableWrapper _int_int_compressed to reset its execution state.
  const auto scan_a = std::make_shared<TableScan>(_int_int_compressed->deep_copy(), greater_than_equals_(column, 4));
  scan_a->set_parameters(parameters);
  EXPECT_EQ(*scan_a->predicate(), *greater_than_equals_(column, 4));

  const auto parameter_expression_with_value = placeholder_(ParameterID{2});
  const auto scan_b = std::make_shared<TableScan>(_int_int_compressed->deep_copy(),
                                                  greater_than_equals_(column, placeholder_(ParameterID{2})));
  scan_b->set_parameters(parameters);
  EXPECT_EQ(*scan_b->predicate(), *greater_than_equals_(column, parameter_expression_with_value));

  const auto scan_c = std::make_shared<TableScan>(_int_int_compressed->deep_copy(),
                                                  greater_than_equals_(column, placeholder_(ParameterID{4})));
  scan_c->set_parameters(parameters);
  EXPECT_EQ(*scan_c->predicate(), *greater_than_equals_(column, placeholder_(ParameterID{4})));
}

TEST_P(OperatorsTableScanTest, GetImpl) {
  /**
   * Test that the correct scanning backend is chosen
   */

  const auto column_a = pqp_column_(ColumnID{0}, DataType::Int, false, "a");
  const auto column_b = pqp_column_(ColumnID{1}, DataType::Float, false, "b");
  const auto column_s = pqp_column_(ColumnID{1}, DataType::String, false, "c");
  const auto column_an = pqp_column_(ColumnID{0}, DataType::String, true, "a");

  // clang-format off
  EXPECT_TRUE(dynamic_cast<ColumnVsValueTableScanImpl*>(TableScan{get_int_float_op(), equals_(column_a, 5)}.create_impl().get()));  // NOLINT
  EXPECT_TRUE(dynamic_cast<ColumnVsValueTableScanImpl*>(TableScan{get_int_float_op(), equals_(5, column_a)}.create_impl().get()));  // NOLINT
  EXPECT_TRUE(dynamic_cast<ColumnVsValueTableScanImpl*>(TableScan{get_int_float_op(), equals_(5, column_a)}.create_impl().get()));  // NOLINT
  EXPECT_TRUE(dynamic_cast<ColumnVsValueTableScanImpl*>(TableScan{get_int_float_op(), equals_(5.0f, column_a)}.create_impl().get()));  // NOLINT
  EXPECT_TRUE(dynamic_cast<ColumnVsValueTableScanImpl*>(TableScan{get_int_float_op(), equals_(5, column_b)}.create_impl().get()));  // NOLINT
  EXPECT_TRUE(dynamic_cast<ColumnVsValueTableScanImpl*>(TableScan{get_int_float_op(), equals_(5.5, column_b)}.create_impl().get()));  // NOLINT
  EXPECT_TRUE(dynamic_cast<ColumnVsValueTableScanImpl*>(TableScan{get_int_float_op(), equals_(5.5f, column_b)}.create_impl().get()));  // NOLINT
  EXPECT_TRUE(dynamic_cast<ColumnBetweenTableScanImpl*>(TableScan{get_int_float_op(), between_inclusive_(column_a, 0, 5)}.create_impl().get()));  // NOLINT
  EXPECT_TRUE(dynamic_cast<ColumnBetweenTableScanImpl*>(TableScan{get_int_float_op(), between_inclusive_(column_a, 0.0f, 5.0)}.create_impl().get()));  // NOLINT
  EXPECT_TRUE(dynamic_cast<ColumnVsColumnTableScanImpl*>(TableScan{get_int_float_op(), equals_(column_b, column_a)}.create_impl().get()));  // NOLINT
  EXPECT_TRUE(dynamic_cast<ColumnLikeTableScanImpl*>(TableScan{get_int_string_op(), like_(column_s, "%s%")}.create_impl().get()));  // NOLINT
  EXPECT_TRUE(dynamic_cast<ExpressionEvaluatorTableScanImpl*>(TableScan{get_int_string_op(), like_("hello", "%s%")}.create_impl().get()));  // NOLINT
  EXPECT_TRUE(dynamic_cast<ExpressionEvaluatorTableScanImpl*>(TableScan{get_int_float_op(), in_(column_a, list_(1, 2, 3))}.create_impl().get()));  // NOLINT
  EXPECT_TRUE(dynamic_cast<ExpressionEvaluatorTableScanImpl*>(TableScan{get_int_float_op(), in_(column_a, list_(1, 2, 3))}.create_impl().get()));  // NOLINT
  EXPECT_TRUE(dynamic_cast<ExpressionEvaluatorTableScanImpl*>(TableScan{get_int_float_op(), and_(greater_than_(column_a, 5), less_than_(column_b, 6))}.create_impl().get()));  // NOLINT
  EXPECT_TRUE(dynamic_cast<ExpressionEvaluatorTableScanImpl*>(TableScan{get_int_float_op(), greater_than_(column_a, 5.5f)}.create_impl().get()));  // NOLINT
  EXPECT_TRUE(dynamic_cast<ExpressionEvaluatorTableScanImpl*>(TableScan{get_int_float_op(), greater_than_(column_b, 1e40)}.create_impl().get()));  // NOLINT
  EXPECT_TRUE(dynamic_cast<ExpressionEvaluatorTableScanImpl*>(TableScan{get_int_float_op(), greater_than_(column_a, int64_t{3'000'000'000})}.create_impl().get()));  // NOLINT
  EXPECT_TRUE(dynamic_cast<ExpressionEvaluatorTableScanImpl*>(TableScan{get_int_float_op(), equals_(column_a, "hello")}.create_impl().get()));  // NOLINT
  EXPECT_TRUE(dynamic_cast<ExpressionEvaluatorTableScanImpl*>(TableScan{get_int_float_op(), between_inclusive_(column_a, 0.1f, 5.5)}.create_impl().get()));  // NOLINT
  EXPECT_TRUE(dynamic_cast<ExpressionEvaluatorTableScanImpl*>(TableScan{get_int_float_op(), between_inclusive_(column_a, 0, null_())}.create_impl().get()));  // NOLINT
  EXPECT_TRUE(dynamic_cast<ColumnIsNullTableScanImpl*>(TableScan{get_int_float_with_null_op(), is_null_(column_an)}.create_impl().get()));  // NOLINT
  EXPECT_TRUE(dynamic_cast<ColumnIsNullTableScanImpl*>(TableScan{get_int_float_with_null_op(), is_not_null_(column_an)}.create_impl().get()));  // NOLINT

  // Cases where the lossless_predicate_cast is used and the predicate condition gets adjusted:
  {
    const auto abstract_impl = TableScan{get_int_float_op(), greater_than_(column_b, 3.1)}.create_impl();
    const auto impl = dynamic_cast<ColumnVsValueTableScanImpl*>(abstract_impl.get());
    ASSERT_TRUE(impl);
    EXPECT_EQ(impl->predicate_condition, PredicateCondition::GreaterThanEquals);
    EXPECT_EQ(impl->value, AllTypeVariant{3.1000001430511474609375f});
  }
  {
    const auto abstract_impl = TableScan{get_int_float_op(), greater_than_equals_(column_b, 3.1)}.create_impl();
    const auto impl = dynamic_cast<ColumnVsValueTableScanImpl*>(abstract_impl.get());
    ASSERT_TRUE(impl);
    EXPECT_EQ(impl->predicate_condition, PredicateCondition::GreaterThanEquals);
    EXPECT_EQ(impl->value, AllTypeVariant{3.1000001430511474609375f});
  }
  {
    const auto abstract_impl = TableScan{get_int_float_op(), equals_(column_b, 3.1)}.create_impl();
    const auto impl = dynamic_cast<ExpressionEvaluatorTableScanImpl*>(abstract_impl.get());
    ASSERT_TRUE(impl);
    // Cannot losslessly convert here.
  }
  {
    const auto abstract_impl = TableScan{get_int_float_op(), less_than_(column_b, 3.1)}.create_impl();
    const auto impl = dynamic_cast<ColumnVsValueTableScanImpl*>(abstract_impl.get());
    ASSERT_TRUE(impl);
    EXPECT_EQ(impl->predicate_condition, PredicateCondition::LessThanEquals);
    EXPECT_EQ(impl->value, AllTypeVariant{3.099999904632568359375f});
  }
  {
    const auto abstract_impl = TableScan{get_int_float_op(), less_than_equals_(column_b, 3.1)}.create_impl();
    const auto impl = dynamic_cast<ColumnVsValueTableScanImpl*>(abstract_impl.get());
    ASSERT_TRUE(impl);
    EXPECT_EQ(impl->predicate_condition, PredicateCondition::LessThanEquals);
    EXPECT_EQ(impl->value, AllTypeVariant{3.099999904632568359375f});
  }

  // Test it once with reversed roles:
  {
    const auto abstract_impl = TableScan{get_int_float_op(), greater_than_(3.1, column_b)}.create_impl();
    const auto impl = dynamic_cast<ColumnVsValueTableScanImpl*>(abstract_impl.get());
    ASSERT_TRUE(impl);
    EXPECT_EQ(impl->predicate_condition, PredicateCondition::LessThanEquals);
    EXPECT_EQ(impl->value, AllTypeVariant{3.099999904632568359375f});
  }

  // The same for different types of between:
  {
    const auto abstract_impl = TableScan{get_int_float_op(), between_inclusive_(column_b, 3.1, 4.1)}.create_impl();
    const auto impl = dynamic_cast<ColumnBetweenTableScanImpl*>(abstract_impl.get());
    ASSERT_TRUE(impl);
    EXPECT_EQ(impl->predicate_condition, PredicateCondition::BetweenInclusive);
    EXPECT_EQ(impl->left_value, AllTypeVariant{3.1000001430511474609375f});
    EXPECT_EQ(impl->right_value, AllTypeVariant{4.099999904632568359375f});
  }

  {
    const auto abstract_impl = TableScan{get_int_float_op(), between_inclusive_(column_b, 3.1, 4.0)}.create_impl();
    const auto impl = dynamic_cast<ColumnBetweenTableScanImpl*>(abstract_impl.get());
    ASSERT_TRUE(impl);
    EXPECT_EQ(impl->predicate_condition, PredicateCondition::BetweenInclusive);
    EXPECT_EQ(impl->left_value, AllTypeVariant{3.1000001430511474609375f});
    EXPECT_EQ(impl->right_value, AllTypeVariant{4.0f});
  }

  {
    const auto abstract_impl = TableScan{get_int_float_op(), between_upper_exclusive_(column_b, 3.1, 4.0)}.create_impl();  // NOLINT
    const auto impl = dynamic_cast<ColumnBetweenTableScanImpl*>(abstract_impl.get());
    ASSERT_TRUE(impl);
    EXPECT_EQ(impl->predicate_condition, PredicateCondition::BetweenUpperExclusive);
    EXPECT_EQ(impl->left_value, AllTypeVariant{3.1000001430511474609375f});
    EXPECT_EQ(impl->right_value, AllTypeVariant{4.0f});
  }

  {
    const auto abstract_impl = TableScan{get_int_float_op(), between_exclusive_(column_b, 3.1, 4.0)}.create_impl();
    const auto impl = dynamic_cast<ColumnBetweenTableScanImpl*>(abstract_impl.get());
    ASSERT_TRUE(impl);
    EXPECT_EQ(impl->predicate_condition, PredicateCondition::BetweenUpperExclusive);
    EXPECT_EQ(impl->left_value, AllTypeVariant{3.1000001430511474609375f});
    EXPECT_EQ(impl->right_value, AllTypeVariant{4.0f});
  }

  {
    const auto abstract_impl = TableScan{get_int_float_op(), between_exclusive_(column_b, 3.1, 4.1)}.create_impl();
    const auto impl = dynamic_cast<ColumnBetweenTableScanImpl*>(abstract_impl.get());
    ASSERT_TRUE(impl);
    EXPECT_EQ(impl->predicate_condition, PredicateCondition::BetweenInclusive);
    EXPECT_EQ(impl->left_value, AllTypeVariant{3.1000001430511474609375f});
    EXPECT_EQ(impl->right_value, AllTypeVariant{4.099999904632568359375f});
  }

  {
    const auto abstract_impl = TableScan{get_int_float_op(), between_exclusive_(column_a, column_b, 3.1)}.create_impl();
    const auto impl = dynamic_cast<ExpressionEvaluatorTableScanImpl*>(abstract_impl.get());
    ASSERT_TRUE(impl);
    // Cannot use ColumnBetweenTableScanImpl as lower bound is not a value
  }

  {
    const auto abstract_impl = TableScan{get_int_float_op(), between_exclusive_(column_a, 3, column_b)}.create_impl();
    const auto impl = dynamic_cast<ExpressionEvaluatorTableScanImpl*>(abstract_impl.get());
    ASSERT_TRUE(impl);
    // Cannot use ColumnBetweenTableScanImpl as upper bound is not a value
  }

  // clang-format on
}

TEST_P(OperatorsTableScanTest, TwoBigScans) {
  // To stress-test the SIMD scan, which only operates on bigger tables, the generated table holds 1'000 rows.
  // For each fifth row, column a is NULL. Otherwise, a is 100'000 + i, b is the index in the list of non-NULL values.

  auto column_definitions = TableColumnDefinitions{{"a", DataType::Int, true}, {"b", DataType::Int, true}};
  const auto data_table = std::make_shared<Table>(column_definitions, TableType::Data, ChunkOffset{13});

  for (auto i = 0, index = 0; i < 1'000; ++i) {
    if (i % 5 == 4) {
      data_table->append({NullValue{}, i});
    } else {
      data_table->append({100'000 + i, index++});
    }
  }

  // We have two full chunks and one open chunk, we only encode the full chunks
  for (auto chunk_id = ChunkID{0}; chunk_id < 2; ++chunk_id) {
    ChunkEncoder::encode_chunk(data_table->get_chunk(chunk_id), {DataType::Int, DataType::Int},
                               {SegmentEncodingSpec{_encoding_type}, SegmentEncodingSpec{EncodingType::Unencoded}});
  }

  auto data_table_wrapper = std::make_shared<TableWrapper>(data_table);
  data_table_wrapper->never_clear_output();
  data_table_wrapper->execute();

  const auto column_a = pqp_column_(ColumnID{0}, DataType::Int, false, "a");

  // Scan for a >= 100'100
  const auto scan_a = std::make_shared<TableScan>(data_table_wrapper, greater_than_equals_(column_a, 100'100));
  scan_a->execute();

  // Now scan for b <= 100'700
  const auto scan_b = std::make_shared<TableScan>(scan_a, less_than_equals_(column_a, 100'700));
  scan_b->execute();

  // Try the same with a between scan
  const auto between_scan =
      std::make_shared<TableScan>(data_table_wrapper, between_inclusive_(column_a, 100'100, 100'700));
  between_scan->execute();

  EXPECT_TABLE_EQ_UNORDERED(scan_b->get_output(), between_scan->get_output());

  const auto& table = scan_b->get_output();
  // Out of the 600 values, 120 are NULL and should have been skipped.
  EXPECT_EQ(table->row_count(), 481);
  for (auto i = 0; i < 481; ++i) {
    EXPECT_EQ(table->get_value<int32_t>(ColumnID{1}, i), 80 + i);
  }
}

/**
 * Tests for sorted_by flag forwarding.
 */
TEST_P(OperatorsTableScanTest, KeepOrderByFlagUnset) {
  // Verify that the sorted_by flag is not set when it's not present in left input.
  const auto scan_unsorted =
      create_table_scan(get_int_float_op(), ColumnID{0}, PredicateCondition::GreaterThanEquals, 1234);
  scan_unsorted->execute();

  const auto& result_table_unsorted = scan_unsorted->get_output();

  EXPECT_TRUE(result_table_unsorted->get_chunk(ChunkID{0})->individually_sorted_by().empty());
  EXPECT_TRUE(result_table_unsorted->get_chunk(ChunkID{1})->individually_sorted_by().empty());
}

TEST_P(OperatorsTableScanTest, SortedFlagDataSegments) {
  // In the current implementation, scans on tables that do not contain ReferenceSegments are executed chunk-by-chunk.
  // Each input chunk will lead to one output chunk (assuming there are qualifying tuples). The position list thus
  // references only a single chunk, allowing the scan to simply forward the initial sorting order.
  const auto scan_sorted =
      create_table_scan(get_int_sorted_op(), ColumnID{0}, PredicateCondition::LessThanEquals, 1234);
  scan_sorted->execute();
  const auto& result_chunk_sorted = scan_sorted->get_output()->get_chunk(ChunkID{0});

  const auto ref_segment =
      std::dynamic_pointer_cast<const ReferenceSegment>(result_chunk_sorted->get_segment(ColumnID{0}));
  EXPECT_TRUE(ref_segment->pos_list()->references_single_chunk());

  const auto expected_sorted_by =
      std::vector<SortColumnDefinition>{SortColumnDefinition(ColumnID{0}, SortMode::Ascending)};
  EXPECT_EQ(result_chunk_sorted->individually_sorted_by(), expected_sorted_by);
}

// Reference segments that only reference a single chunk allow forwarding the sorting flags.
TEST_P(OperatorsTableScanTest, SortedFlagReferenceSegments) {
  const auto table = get_int_sorted_op()->table;

  const auto ref_table = std::make_shared<Table>(table->column_definitions(), TableType::References);

  auto pos_list_1 = std::make_shared<RowIDPosList>();
  pos_list_1->emplace_back(ChunkID{0}, ChunkOffset{1});
  pos_list_1->emplace_back(ChunkID{0}, ChunkOffset{0});
  pos_list_1->emplace_back(ChunkID{0}, ChunkOffset{3});
  pos_list_1->emplace_back(ChunkID{0}, ChunkOffset{2});
  pos_list_1->guarantee_single_chunk();

  const auto segment_1 = std::make_shared<ReferenceSegment>(table, ColumnID{0}, pos_list_1);
  ref_table->append_chunk({segment_1});

  auto pos_list_2 = std::make_shared<RowIDPosList>();
  pos_list_2->emplace_back(ChunkID{1}, ChunkOffset{0});
  pos_list_2->guarantee_single_chunk();

  const auto segment_2 = std::make_shared<ReferenceSegment>(table, ColumnID{0}, pos_list_2);
  ref_table->append_chunk({segment_2});

  ref_table->get_chunk(ChunkID{0})->finalize();
  ref_table->get_chunk(ChunkID{1})->finalize();
  ref_table->get_chunk(ChunkID{0})->set_individually_sorted_by(SortColumnDefinition(ColumnID{0}, SortMode::Ascending));
  ref_table->get_chunk(ChunkID{1})->set_individually_sorted_by(SortColumnDefinition(ColumnID{0}, SortMode::Ascending));
  auto table_wrapper = std::make_shared<TableWrapper>(std::move(ref_table));
  table_wrapper->execute();

  const auto scan_sorted = create_table_scan(table_wrapper, ColumnID{0}, PredicateCondition::LessThanEquals, 4);
  scan_sorted->execute();

  EXPECT_EQ(scan_sorted->get_output()->chunk_count(), 2);
  for (const auto& chunk_id : {ChunkID{0}, ChunkID{1}}) {
    const auto& result_chunk_sorted = scan_sorted->get_output()->get_chunk(chunk_id);

    const auto ref_segment =
        std::dynamic_pointer_cast<const ReferenceSegment>(result_chunk_sorted->get_segment(ColumnID{0}));
    EXPECT_TRUE(ref_segment->pos_list()->references_single_chunk());

    const auto& chunk_sorted_by = result_chunk_sorted->individually_sorted_by();
    ASSERT_EQ(chunk_sorted_by, std::vector{SortColumnDefinition(ColumnID{0}, SortMode::Ascending)});
  }
}

// Scanning two sorted reference segments. Currently, the sorting information is not passed forward when reference
// segments do not guarantee that they only reference a single physical chunk (see table_scan.cpp). An exception is
// if the entire chunk matches and is simply forwarded.
TEST_P(OperatorsTableScanTest, SortedFlagSingleChunkNotGuaranteed) {
  const auto table = get_int_sorted_op()->table;

  const auto ref_table = std::make_shared<Table>(table->column_definitions(), TableType::References);

  auto pos_list_1 = std::make_shared<RowIDPosList>();
  pos_list_1->emplace_back(ChunkID{0}, ChunkOffset{1});
  pos_list_1->emplace_back(ChunkID{0}, ChunkOffset{3});

  const auto segment_1 = std::make_shared<ReferenceSegment>(table, ColumnID{0}, pos_list_1);
  ref_table->append_chunk({segment_1});

  auto pos_list_2 = std::make_shared<RowIDPosList>();
  pos_list_2->emplace_back(ChunkID{1}, ChunkOffset{0});

  const auto segment_2 = std::make_shared<ReferenceSegment>(table, ColumnID{0}, pos_list_2);
  ref_table->append_chunk({segment_2});

  ref_table->get_chunk(ChunkID{0})->finalize();
  ref_table->get_chunk(ChunkID{1})->finalize();
  ref_table->get_chunk(ChunkID{0})->set_individually_sorted_by(SortColumnDefinition(ColumnID{0}, SortMode::Ascending));
  ref_table->get_chunk(ChunkID{1})->set_individually_sorted_by(SortColumnDefinition(ColumnID{0}, SortMode::Ascending));
  auto table_wrapper = std::make_shared<TableWrapper>(std::move(ref_table));
  table_wrapper->execute();

  const auto scan_sorted = create_table_scan(table_wrapper, ColumnID{0}, PredicateCondition::GreaterThanEquals, 2);
  scan_sorted->execute();

  EXPECT_EQ(scan_sorted->get_output()->chunk_count(), 2);

  {
    // Chunk 0 matches partially and no sorting information is forwarded.
    const auto& result_chunk_sorted = scan_sorted->get_output()->get_chunk(ChunkID{0});
    const auto& chunk_sorted_by = result_chunk_sorted->individually_sorted_by();
    ASSERT_TRUE(chunk_sorted_by.empty());
  }

  {
    // Chunk 1 matches entirely and the chunk (including the sorting information) is forwarded.
    const auto& result_chunk_sorted = scan_sorted->get_output()->get_chunk(ChunkID{1});
    const auto& chunk_sorted_by = result_chunk_sorted->individually_sorted_by();
    ASSERT_EQ(chunk_sorted_by.size(), 1);
    ASSERT_EQ(chunk_sorted_by.at(0), SortColumnDefinition(ColumnID(0), SortMode::Ascending));
  }
}

// Tables with ReferenceSegments that reference multiple chunks are first separated by the ChunkID (see
// AbstractDereferencedColumnTableScanImpl::_scan_reference_segment) and thus the order may be destroyed. Due to the
// missing single chunk guarantee, no sorting information is forwarded. An exception is if the entire chunk matches and
// is simply forwarded. This is NOT tested here.
TEST_P(OperatorsTableScanTest, SortedFlagMultipleChunksReferenced) {
  const auto table = get_int_sorted_op()->table;

  const auto ref_table = std::make_shared<Table>(table->column_definitions(), TableType::References);

  auto pos_list_1 = std::make_shared<RowIDPosList>();
  pos_list_1->emplace_back(ChunkID{0}, ChunkOffset{1});
  pos_list_1->emplace_back(ChunkID{0}, ChunkOffset{0});
  pos_list_1->emplace_back(ChunkID{0}, ChunkOffset{3});
  pos_list_1->emplace_back(ChunkID{0}, ChunkOffset{2});
  pos_list_1->emplace_back(ChunkID{1}, ChunkOffset{0});

  const auto segment = std::make_shared<ReferenceSegment>(table, ColumnID{0}, pos_list_1);
  ref_table->append_chunk({segment});

  ref_table->get_chunk(ChunkID{0})->finalize();
  ref_table->get_chunk(ChunkID{0})->set_individually_sorted_by(SortColumnDefinition(ColumnID{0}, SortMode::Ascending));
  auto table_wrapper = std::make_shared<TableWrapper>(std::move(ref_table));
  table_wrapper->execute();

  const auto scan_sorted = create_table_scan(table_wrapper, ColumnID{0}, PredicateCondition::GreaterThanEquals, 2);
  scan_sorted->execute();

  EXPECT_EQ(scan_sorted->get_output()->chunk_count(), 1);
  const auto& result_chunk_sorted = scan_sorted->get_output()->get_chunk(ChunkID{0});
  const auto& chunk_sorted_by = result_chunk_sorted->individually_sorted_by();
  ASSERT_TRUE(chunk_sorted_by.empty());
}

TEST_P(OperatorsTableScanTest, DeepCopyRetainsExcludedChunks) {
  const auto table_scan =
      create_table_scan(get_int_float_op(), ColumnID{0}, PredicateCondition::GreaterThanEquals, 1234);
  table_scan->excluded_chunk_ids = std::make_shared<std::vector<ChunkID>>(std::initializer_list<ChunkID>{ChunkID{0}});
  const auto new_table_scan = std::dynamic_pointer_cast<TableScan>(table_scan->deep_copy());
  EXPECT_EQ(*table_scan->excluded_chunk_ids, *new_table_scan->excluded_chunk_ids);
  EXPECT_EQ(table_scan->excluded_chunk_ids->data(),
            new_table_scan->excluded_chunk_ids->data());  // Should be the same object.
}

}  // namespace hyrise
